package com.practice.gmall.realtime.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

public class FlinkSourceUtil {
    public static KafkaSource<String> getKafkaSource(String kafkaBrokers, String topic, String groupId) {
        KafkaSource<String> source = KafkaSource.<String>builder().setBootstrapServers(kafkaBrokers)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                // .setStartingOffsets(OffsetsInitializer.committedOffsets())// set offset as the group commited offset
                .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")// ensure exactly-once semantic
                // todo Implement interface KafkaDeserializationSchema, this deserializer will case bug later
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        return source;
    }
}
